…#2465)

# Which issue does this PR close?

Part of #1863.

#1863 enumerates all three argument forms of `UNIX_TIMESTAMP`. The
1-argument and 2-argument string-parsing forms landed in #2409 and
#2448, and this PR adds the remaining 0-argument form. #1863 stays open
after this one merges, for the timezone configuration work tracked as
sub-issue #2455.

# Rationale for this change

`UNIX_TIMESTAMP()` with no arguments returns the current epoch seconds.
The converter rejected it, so any Calc containing it fell back to Flink
wholesale, taking every other expression in that Calc down with it.

Two things were believed to block it. Both turned out to be false, and I
stated one of them myself earlier in this issue.

**"A zero-argument ext function cannot size its output."** The
ext-function ABI drops the batch row count, so a 0-argument function
appears to have no way to know how many values to return. That is true
only for a function returning an `Array`. The length check in
`ScalarFunctionExpr::evaluate` sits entirely inside the
`ColumnarValue::Array` branch, so a function returning
`ColumnarValue::Scalar` never reaches it, and the projection broadcasts
the scalar to the batch width. No operand is needed, and no carrier
column either.

**"The 0-argument form needs the session time zone."** It does not. The
generated Flink operator holds a `timeZone` field, passes it to the
1-argument and 2-argument calls, and emits `result$1 =
DateTimeUtils.unixTimestamp()` with no argument at all. The result is
epoch seconds, which is zone-independent.

An approach was proposed on #1863 that works around the first premise:
pass any input column to the Rust function purely to convey the batch
length, ignore its data, and return that many timestamps. That would
have worked, and it was a sound answer to the constraint as I had
described it. This PR does not use it, because once the constraint turns
out not to exist the carrier column has no job left. The difference is
visible in the diff, so it is worth stating plainly: no column is
passed, the Rust function reads no length, and the row count comes from
the projection broadcasting a scalar. The observable behavior is the
same either way.

# What changes are included in this PR?

Three commits.

1. A native zero-argument ext function `Flink_UnixTimestampNow`
returning `ColumnarValue::Scalar(Int64)`. The clock is read as
`timestamp_millis() / 1000`, mirroring Flink's
`System.currentTimeMillis() / 1000` operator for operator.
2. The converter admits the 0-argument form and emits that function with
an empty operand list. The 0-argument arm sits **above** the session
time zone check, since a zone the native side cannot resolve is no
reason to fall back a query that never consults a zone. Class and method
javadoc are corrected: they previously described the 0-argument form as
falling back and described `UNIX_TIMESTAMP` as always mapping to
`Flink_UnixTimestamp` with `[value, chronoFormat, zoneId]`.
3. End-to-end coverage in `AuronFlinkCalcITCase`.

No protobuf change, no `Cargo.toml` change, no `pom.xml` change. It is a
separate native function rather than a fourth arity on
`Flink_UnixTimestamp`, whose contract is "parse this string with this
format in this zone" and which the 0-argument form shares none of.

# Are there any user-facing changes?

Yes, and one of them is a deliberate semantic divergence worth reviewing
on its own merits.

A query using `UNIX_TIMESTAMP()` that previously fell back to Flink now
runs natively.

**Flink evaluates the niladic form per record. This implementation reads
the clock once per call site per evaluation and broadcasts it across the
batch**, so rows in one batch share a timestamp, and that timestamp is
taken when the batch is evaluated rather than when each row arrived. Two
`UNIX_TIMESTAMP()` calls in one projection are read independently, which
matches Flink.

How large the skew can get depends on what closes the batch:

| Path | Bound |
|---|---|
| Calc with a declared `WATERMARK FOR` | about 205 ms, the default
`autoWatermarkInterval` |
| Calc with no watermark strategy | 8192 rows only, so `8192 / rate`
seconds. About 82 s at 100 rec/s |
| Fused Kafka plan | the native scan blocks until its buffer fills,
default 3000, with no time flush |

The unwatermarked case is the common shape for a processing-time query,
so this is not a corner case. It was discussed and settled on the issue
(#1863 (comment)):
per-batch is acceptable, since the zero-argument form is non-idempotent
anyway and what users generally want is to know roughly when a record
was processed. It is also the finest granularity among comparable
engines: StreamFusion stamps `PROCTIME()` once per operator compile,
Flink's own batch mode folds `CURRENT_TIMESTAMP` to a query-start
literal, DataFusion, ClickHouse and Velox are per query, DuckDB is per
transaction, and ClickHouse ships `nowInBlock()` as a documented
per-block clock. Flink also declares the niladic form only
`SqlMonotonicity.INCREASING`, which a per-batch constant satisfies.

Gating it conditionally on a declared watermark was considered and
rejected: the converter has no access to the ExecNode graph, the
predicate misses DataStream-level watermarks,
`auto-watermark-interval=0` and idle partitions, it keys native support
off an unrelated DDL clause, and it only shrinks the skew rather than
removing it.

# How was this patch tested?

Native unit tests for the scalar return and its broadcast to batch
width, and for the value being a plausible epoch second rather than
milliseconds.

Converter tests for the emitted node shape, and
`testUnixTimestampZeroArgSupportedWithFixedOffsetZone`, which exists to
catch the 0-argument arm being placed below the time zone check. It is a
real discriminator: moving the arm below the check fails that test and
no other. Two tests that asserted the form falls back are inverted.

An ITCase asserting one clock-bracketed row per input row and no
recorded fallback. Both assertions are load-bearing and neither subsumes
the other. A zero fallback count establishes that the Calc converted,
not that it ran: a native library holding no registry arm for the
function still converts at plan time and fails only during execution,
where nothing records a fallback, leaving the counter at zero and the
result set empty. The row count is what establishes the native plan
executed.

The ITCase was checked to be non-vacuous by running rather than by
inspection: reverting the converter gate fails it on the fallback count,
and running against a library without the registry arm fails it on the
row count.

Full module build green, 0 checkstyle violations, spotless clean.

# Was this patch authored or co-authored using generative AI tooling?
- [x] Yes
- [ ] No

`Generated-by: Claude Code (Claude Opus 5)`